package com.dempe.forest.common.discovery;

import com.dempe.forest.common.utils.Utils;
import com.google.common.collect.Lists;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.Participant;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.util.Collection;
import java.util.List;

/**
 * 进程名称服务
 * User: Dempe
 * Date: 2016/1/29
 * Time: 16:36
 * To change this template use File | Settings | File Templates.
 */
public class DiscoveryService implements NameService, Closeable {

    private final static Logger LOGGER = LoggerFactory.getLogger(DiscoveryService.class);

    private String path = "/forest/discovery";

    private String connStr = "120.76.129.232:2181";


    private String ip;
    private int port;
    private String name;

    private String id;

    private ServiceDiscovery<NodeDetails> serviceDiscovery;

    private CuratorFramework zkClient;

    private DiscoveryServerService discoveryServerService;

    private LeaderLatch leaderLatch;


    public DiscoveryService(String ip, int port, String name, String path, String connStr) throws Exception {
        this.ip = ip;
        this.name = name;
        this.port = port;
        this.path = path;
        this.connStr = connStr;
        id = Utils.md5Str(ip + "|" + port + "|" + name);
        init();
        addCloseShutDownHook();
    }

    public DiscoveryService() throws Exception {
        id = Utils.md5Str(ip + "|" + port + "|" + name);
        init();
        addCloseShutDownHook();
    }


    public void start() throws Exception {
        this.serviceDiscovery.start();
    }

    public void init() throws Exception {
        this.zkClient = CuratorFrameworkFactory.newClient(connStr, new ExponentialBackoffRetry(1000, 3));
        zkClient.start();
        JsonInstanceSerializer<NodeDetails> serializer = new JsonInstanceSerializer<NodeDetails>(NodeDetails.class);
        this.serviceDiscovery = ServiceDiscoveryBuilder.builder(NodeDetails.class).client(zkClient).basePath(path).serializer(serializer).build();
        leaderLatch = new LeaderLatch(zkClient, path, id);
        register();

    }


    @Override
    public boolean hasLeadership() {
        return leaderLatch.hasLeadership();
    }

    @Override
    public void register() throws Exception {
        NodeDetails nodeDetails = new NodeDetails();
        nodeDetails.setName(name);
        nodeDetails.setIp(ip);
        nodeDetails.setPort(port);
        nodeDetails.setId(id);
        discoveryServerService = new DiscoveryServerService(zkClient, path, nodeDetails);
        discoveryServerService.start();
        leaderLatch.start();
    }

    public Collection<Participant> getParticipants() throws Exception {
        return leaderLatch.getParticipants();
    }

    @Override
    public List<Collection<ServiceInstance<NodeDetails>>> list() throws Exception {
        Collection<String> serviceNames = serviceDiscovery.queryForNames();
        List<Collection<ServiceInstance<NodeDetails>>> list = Lists.newArrayList();
        for (String serviceName : serviceNames) {
            Collection<ServiceInstance<NodeDetails>> instances = serviceDiscovery.queryForInstances(serviceName);
            list.add(instances);
        }
        return list;
    }

    @Override
    public Collection<ServiceInstance<NodeDetails>> listByName(String name) throws Exception {
        return serviceDiscovery.queryForInstances(name);
    }

    public void addCloseShutDownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    close();
                } catch (IOException e) {
                    LOGGER.error(e.getMessage(), e);
                }
            }
        }));
    }

    public void close() throws IOException {
        if (discoveryServerService != null) {
            discoveryServerService.close();
        }
        if (serviceDiscovery != null) {
            serviceDiscovery.close();
        }

    }

    public String getPath() {
        return path;
    }

    public void setPath(String path) {
        this.path = path;
    }

    public String getConnStr() {
        return connStr;
    }

    public void setConnStr(String connStr) {
        this.connStr = connStr;
    }

    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }
}
